package realtime.util;/*
 *@ClassName : ClickHouseUtil
 *@Description : $ {description}
 *@Author : NingYu
 *@Date : 2023-03-24 11:{MINUTE}
 */

import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import realtime.bean.TransientSink;
import realtime.common.GmallConfig;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class ClickHouseUtil {

    //sql:insert into db.t values(?,?,?,?,?)
    public static <T> SinkFunction<T> getSinkFunction(String sql){
        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        //反射 获取所有字段
                        Class<?> clz = t.getClass();
                        Field[] fields = clz.getDeclaredFields();

                        int offset = 0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];
                            field.setAccessible(true);//public或private都能访问
                            //todo 尝试获取字段的标记，如果标记纯在，则跳过该字段
                            TransientSink sink = field.getAnnotation(TransientSink.class);
                            if (sink != null){
                                offset ++;
                                continue;
                            }
                            Object value = field.get(t);
                            preparedStatement.setObject((i + 1) - offset,value);
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(1000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
    }

}